#if !defined(LOG_TAG)
    #define LOG_TAG                    "threadpool"
#endif
#if defined(THREADPOOL_LOG_LVL)
    #define LOG_LVL                    THREADPOOL_LOG_LVL
#endif

#include "threadpool.h"

#define THREADPOOL_VER_MAJOR  (1)
#define THREADPOOL_VER_MINOR  (0)
#define THREADPOOL_VER_PATCH  (0)

// 工作线程函数
static void* threadpool_worker_thread(void* context) 
{
    struct threadpool* pthreadpool = (struct threadpool*)context;

    // log_d("threadpool_worker_thread start");

    while (1) {
        // 等待任务或关闭信号
        while (list_empty(&pthreadpool->task_busy_queue) &&!pthreadpool->shutdown) {
            wait_complete(&pthreadpool->wait);
        }

        // 处理关闭信号
        if (pthreadpool->shutdown) {
            break;
        }

        // 提取任务
        struct thread_pool_task* ptask = NULL;

        ptask = pthreadpool->op->get_task(pthreadpool, &pthreadpool->task_busy_queue);
        if (ptask != NULL) {
            // 执行任务
            lock(&pthreadpool->lock);
            pthreadpool->active_count++;
            unlock(&pthreadpool->lock);

            ptask->function(ptask->arg);

            lock(&pthreadpool->lock);
            pthreadpool->active_count--;
            unlock(&pthreadpool->lock);

            pthreadpool->op->put_task(pthreadpool, &pthreadpool->task_free_queue, ptask);
        }
    }

    return NULL;
}

// 添加任务
static int32_t threadpool_add_task(struct threadpool* pthreadpool, void (*function)(void*), void* arg) 
{
    int32_t ret = 0;

    if (!pthreadpool || !function) {
        return -1;
    }

    if (pthreadpool->shutdown) {
        return -1;
    }

    struct thread_pool_task* ptask = NULL;
    ptask = pthreadpool->op->get_task(pthreadpool, &pthreadpool->task_free_queue);
    if (ptask != NULL) {
        ptask->function = function;
        ptask->arg = arg;
        
        pthreadpool->op->put_task(pthreadpool, &pthreadpool->task_busy_queue, ptask);
    }

    wait_wakeup(&pthreadpool->wait);

    return ret;
}

// 销毁未执行任务
static int32_t threadpool_drain_tasks(struct threadpool* pthreadpool) 
{
    int32_t ret = 0;

    if (!pthreadpool) {
        return -1;
    }

    struct thread_pool_task* ptask = NULL;

    while(!list_empty(&pthreadpool->task_busy_queue)){
        ptask = pthreadpool->op->get_task(pthreadpool, &pthreadpool->task_busy_queue);
        if (ptask != NULL) {
            pthreadpool->op->put_task(pthreadpool, &pthreadpool->task_free_queue, ptask);
        }
    }

    return ret;
}

// 获取运行中任务数
static int32_t threadpool_get_busy_task_count(struct threadpool* pthreadpool)
{
    if (!pthreadpool) {
        return -1;
    }

    lock(&pthreadpool->lock);
    int32_t count = pthreadpool->active_count;
    unlock(&pthreadpool->lock);
    
    return count;
}

// 获取空闲线程数
static int32_t threadpool_get_idle_thread_count(struct threadpool* pthreadpool)
{
    if (!pthreadpool) {
        return -1;
    }

    lock(&pthreadpool->lock);
    int32_t idle = pthreadpool->num_threads - pthreadpool->active_count;
    unlock(&pthreadpool->lock);

    return idle;
}

static int32_t threadpool_start(struct threadpool* pthreadpool, struct threadpool_config *pcfg)
{
    int32_t ret = 0;

    if(!pthreadpool ||!pcfg){
        return -EINVAL;
    }

    uint32_t num_threads = 4;//pcfg->num_threads;
    if(num_threads == 0){
        return -EINVAL;
    }

    pthreadpool->pcfg = pcfg;
    pthreadpool->num_threads = num_threads;
    pthreadpool->shutdown = false;

    pthreadpool->workers = malloc(sizeof(pthread_t) * pthreadpool->num_threads);
    if (!pthreadpool->workers) {
        return -ENOMEM;
    }

    for (int i = 0; i < pthreadpool->num_threads; i++) {
        if (pthread_create(&pthreadpool->workers[i], NULL, threadpool_worker_thread, pthreadpool) != 0) {
            return -1;
        }
    }

    struct thread_pool_task* ptask = NULL;
    for (int i = 0; i < THREADPOOL_MAX_TASK_NUM; i++) {
        ptask = &pthreadpool->task[i];
        ptask->function = NULL;
        ptask->arg = NULL;
        pthreadpool->op->put_task(pthreadpool, &pthreadpool->task_free_queue, ptask);
    }

    pthreadpool->enable = true;

    log_i("threadpool start successful.");
    
    return ret;
}

static int32_t threadpool_stop(struct threadpool* pthreadpool)
{
    int32_t ret = 0;

    lock(&pthreadpool->lock);
    pthreadpool->shutdown = true;
    wait_cancel(&pthreadpool->wait);
    unlock(&pthreadpool->lock);

    for (int i = 0; i < pthreadpool->num_threads; i++) {
        pthread_join(pthreadpool->workers[i], NULL);
    }

    if(pthreadpool->workers){
        free(pthreadpool->workers);
        pthreadpool->workers = NULL;
    }

    pthreadpool->enable = false;

    log_i("threadpool stop successful.");

    return ret;
}

static int32_t threadpool_init(struct threadpool* pthreadpool)
{
    int32_t ret = 0;

    log_i("THREADPOOL VERSION:V%d.%d.%d [%s %s]", THREADPOOL_VER_MAJOR, THREADPOOL_VER_MINOR, THREADPOOL_VER_PATCH, __DATE__, __TIME__);

    INIT_LIST_HEAD(&(pthreadpool->head));
    lock_init(&(pthreadpool->lock));
    lock_init(&(pthreadpool->notify_lock));
    wait_init(&(pthreadpool->wait));

    INIT_LIST_HEAD(&(pthreadpool->task_free_queue));
    INIT_LIST_HEAD(&(pthreadpool->task_busy_queue));
    
    return ret;
}

static int32_t threadpool_release(struct threadpool* pthreadpool)
{
    pthreadpool->op->stop(pthreadpool);

    lock_destroy((&pthreadpool->notify_lock));
    lock_destroy((&pthreadpool->lock));
    wait_destroy((&pthreadpool->wait));

    return 0;
}

static int32_t threadpool_put_task(struct threadpool* pthreadpool, struct list_head* queue, struct thread_pool_task* ptask)
{
	lock(&pthreadpool->lock);
	list_add_tail(&ptask->head, queue);
    queue->size++;
	unlock(&pthreadpool->lock);
	return 0;
}

static struct thread_pool_task* threadpool_get_task(struct threadpool* pthreadpool, struct list_head* queue)
{
	struct thread_pool_task* ptask = NULL;

	lock(&pthreadpool->lock);
	if(!list_empty(queue)){
		ptask = list_first_entry(queue, struct thread_pool_task, head);
		list_del(&ptask->head);
        queue->size--;
	}
	unlock(&pthreadpool->lock);

	return ptask;
}

static int32_t threadpool_status(struct threadpool* pthreadpool)
{
    int32_t nums = 0;

    lock(&pthreadpool->lock);
    pthreadpool->idle_count   = pthreadpool->num_threads - pthreadpool->active_count;
    unlock(&pthreadpool->lock);

    log_d("threadpool status: active_count:%d, idle_count:%d busy_count:%d free_count:%d",
           pthreadpool->active_count, pthreadpool->idle_count,
           pthreadpool->task_busy_queue.size, pthreadpool->task_free_queue.size);

    return nums;
}

static int32_t threadpool_register_notify(struct threadpool* pthreadpool, int32_t event, threadpool_event_notify notify, void* object)
{
    struct threadpool_event_action* paction;
    if(!notify || (event <= THREADPOOL_EVENT_NONE) || (event >= THREADPOOL_EVENT_MAX)){
        return -EINVAL;
    }
    paction = (struct threadpool_event_action*)malloc(sizeof(struct threadpool_event_action));
    if(!paction){
        return -ENOMEM;
    }
    paction->notify = notify;
    paction->object = object;
    lock(&(pthreadpool->notify_lock));
    paction->next = pthreadpool->paction[event];
    pthreadpool->paction[event] = paction;
    unlock(&(pthreadpool->notify_lock));
    return 0;
}

static int32_t threadpool_unregister_notify(struct threadpool* pthreadpool, int32_t event, void* object)
{
    struct threadpool_event_action *paction,* ptmp;
    if((event <= THREADPOOL_EVENT_NONE) || (event >= THREADPOOL_EVENT_MAX)){
        return -EINVAL;
    }
    lock(&(pthreadpool->notify_lock));
    paction = pthreadpool->paction[event];
    if(paction){
        if(paction->object == object){
            pthreadpool->paction[event] = paction->next;
            free(paction);
        }else{
            while(paction->next){
                if(paction->next->object == object){
                    ptmp = paction->next;
                    paction->next = ptmp->next;
                    free(ptmp);
                    break;
                }
                paction = paction->next;
            }
        }
    }
    unlock(&(pthreadpool->notify_lock));

    return 0;
}

static int32_t threadpool_trigger_notify(struct threadpool* pthreadpool, int32_t event, void* context)
{
    struct threadpool_event_action* paction;
    if((event <= THREADPOOL_EVENT_NONE) || (event >= THREADPOOL_EVENT_MAX)){
        return -EINVAL;
    }
    lock(&(pthreadpool->notify_lock));
    paction = pthreadpool->paction[event];
    while(paction){
        paction->notify(pthreadpool, event, paction->object, context);
        paction = paction->next;
    }
    unlock(&(pthreadpool->notify_lock));

    return 0;
}

static struct threadpool_operation threadpool_op =
{
    .init = threadpool_init,
    .release = threadpool_release,

    .start = threadpool_start,
    .stop = threadpool_stop,

    .add_task = threadpool_add_task,
    .drain_tasks = threadpool_drain_tasks,
    .status = threadpool_status,

    .put_task = threadpool_put_task,
    .get_task = threadpool_get_task,

    .register_notify = threadpool_register_notify,
    .unregister_notify = threadpool_unregister_notify,
    .trigger_notify = threadpool_trigger_notify,
};

int32_t create_init_threadpool(struct threadpool** pthreadpool)
{
    int32_t ret;
    struct threadpool* ptmp;
    (*pthreadpool) = (struct threadpool*)malloc(sizeof(struct threadpool));
    if(!(*pthreadpool)){
        return -ENOMEM;
    }
    ptmp = *pthreadpool;
    memset(ptmp, 0, sizeof(struct threadpool));
    ptmp->op = &threadpool_op;
    ret = ptmp->op->init(ptmp);
    if(ret < 0){
        release_destroy_threadpool(ptmp);
        return ret;
    }
    return 0;
}

void release_destroy_threadpool(struct threadpool* pthreadpool)
{
    if(pthreadpool){
        pthreadpool->op->release(pthreadpool);
        free(pthreadpool);
    }
}
        